package data_join;

import Utils.FileUtil;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.*;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.List;

/*
new sort of data
Orders
id gid amount
001 01 2
002 02 2
003 03 4
004 01 1
005 02 3
006 03 2
007 04 2
008 07 1
009 06 2
010 05 3
011 07 4
012 05 3
013 04 2
014 01 3
015 06 2
016 06 9
017 06 4

Items
01 computer
02 refrigerator
03 TV
04 watch
05 microwave
06 vacuum cleaner
07 washing machine

 */
public class ReduceJoin {

    public static String inputOrderPath = "hdfs://localhost:9000/join/join_order";
    public static String inputItemPath = "hdfs://localhost:9000/join/join_item";
    public static String outputPath = "hdfs://localhost:9000/join/output_orderitem";

    public static void main(String[] args)
            throws IOException, InterruptedException, ClassNotFoundException {
        Configuration conf = new Configuration();
        FileUtil fileUtil = new FileUtil(conf, outputPath);
        fileUtil.doError(inputOrderPath);
        fileUtil.doError(inputItemPath);
        fileUtil.doDelete(outputPath);

        Job job = Job.getInstance(conf, "reduce-end join");
        job.setJarByClass(MapJoin.class);

        job.setMapperClass(doMapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(GoodsBean.class);

        job.setReducerClass(doReducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);

        //job.setGroupingComparatorClass(doComparator.class);
        FileInputFormat.setInputPaths(job, new Path(inputItemPath), new Path(inputOrderPath));
        FileOutputFormat.setOutputPath(job, new Path(outputPath));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }


    public static class doMapper extends Mapper<LongWritable, Text, Text, GoodsBean> {
        private GoodsBean bean = new GoodsBean();
        private String fileName;
        private Text gid = new Text();

        @Override
        protected void setup(Mapper<LongWritable, Text, Text, GoodsBean>.Context context)
                throws IOException, InterruptedException {
            // due to the fact that two tables are stored in two separated
            // files, we need to first specify the name of each file
            FileSplit inputSplit = (FileSplit) context.getInputSplit();
            fileName = inputSplit.getPath().getName();
        }

        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, GoodsBean>.Context context)
                throws IOException, InterruptedException {
            String[] fields = value.toString().trim().split(" ");
            // Orders table : oid gid amount
            // Items table : gid name
            if (fileName.equals("join_order")) {
                bean.setOid(fields[0]);
                bean.setGid(fields[1]);
                bean.setAmount(Integer.parseInt(fields[2]));
                bean.setName("");
                bean.setTable("order");
                gid.set(fields[1]);
            } else {
                bean.setOid("");
                bean.setGid(fields[0]);
                bean.setAmount(0);
                bean.setName(fields[1]);
                bean.setTable("item");
                gid.set(fields[0]);
            }
            //System.out.println(bean.toString());
            context.write(gid, bean);
        }
    }

    public static class doReducer extends Reducer<Text, GoodsBean, Text, NullWritable> {

        private List<GoodsBean> orders = new ArrayList<>();
        private GoodsBean itBean = new GoodsBean();
        private Text result = new Text();

        @Override
        protected void reduce(Text key, Iterable<GoodsBean> values, Reducer<Text, GoodsBean, Text, NullWritable>.Context context)
                throws IOException, InterruptedException {
            for (GoodsBean bean : values) {
                //System.out.println(bean.toString());
                if (bean.getTable().equals("order")) {
                    try {
                        GoodsBean currentBean = new GoodsBean();
                        BeanUtils.copyProperties(currentBean, bean);
                        orders.add(currentBean);
                    } catch (IllegalAccessException e) {
                        e.printStackTrace();
                    } catch (InvocationTargetException e) {
                        e.printStackTrace();
                    }
                } else {
                    try {
                        BeanUtils.copyProperties(itBean, bean);
                    } catch (IllegalAccessException e) {
                        e.printStackTrace();
                    } catch (InvocationTargetException e) {
                        e.printStackTrace();
                    }
                }
            }

            for (GoodsBean bean : orders) {
                bean.setName((itBean.getName()));
                result.set(bean.toString());
                context.write(result, NullWritable.get());
            }

        }
    }

    public static class doComparator extends WritableComparator {

        public doComparator() {
            super(GoodsBean.class, true);
        }

        @Override
        public int compare(WritableComparable a, WritableComparable b) {
            GoodsBean oa = (GoodsBean) a;
            GoodsBean ob = (GoodsBean) b;
            return oa.getGid().compareTo(ob.getGid());
        }
    }
}

